package api.flinkapi.commons;

import org.apache.commons.lang.StringUtils;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.util.Random;

public class RandomTempSource extends RichSourceFunction<String> {
    Random rd = new Random();
    int interval;
    volatile boolean flag = true;

    public RandomTempSource() {
        this.interval = 1000;
    }

    public RandomTempSource(int interval) {
        this.interval = interval;
    }


    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while(flag){
            double temp = rd.nextGaussian()+60;
            long ts = System.currentTimeMillis();
            String detectorId = StringUtils.leftPad(rd.nextInt(10) + "", 2, "0");


            double wet = rd.nextGaussian()+90;

            ctx.collect(detectorId+","+ts+","+temp+","+wet);

            Thread.sleep(interval);

        }
    }

    @Override
    public void cancel() {

        flag = true;
    }
}
